/**
 * Licensed to the Apache Software Foundation (ASF) under one
 * or more contributor license agreements.  See the NOTICE file
 * distributed with this work for additional information
 * regarding copyright ownership.  The ASF licenses this file
 * to you under the Apache License, Version 2.0 (the
 * "License"); you may not use this file except in compliance
 * with the License.  You may obtain a copy of the License at
 *
 *     http://www.apache.org/licenses/LICENSE-2.0
 *
 * Unless required by applicable law or agreed to in writing, software
 * distributed under the License is distributed on an "AS IS" BASIS,
 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 * See the License for the specific language governing permissions and
 * limitations under the License.
 */
package com.nhncorp.neptune.changelog;

import java.io.DataInputStream;
import java.io.DataOutputStream;
import java.io.IOException;
import java.lang.reflect.Method;
import java.net.InetSocketAddress;
import java.net.Socket;
import java.net.UnknownHostException;
import java.nio.ByteBuffer;
import java.nio.channels.SocketChannel;
import java.util.Arrays;

import com.nhncorp.neptune.common.conf.NConfiguration;
import com.nhncorp.neptune.common.ipc.NRPC;

public class TestClient {
  private static final int TRY = 3;
  static InetSocketAddress[] rpcAddresses;

  public static void main(String[] args) {
    SocketChannel sc = null;
    try {
      sc = SocketChannel.open();
      sc.connect(new InetSocketAddress("127.0.0.1", 18000));
      if (connectPipeTo(sc.socket())) {
        for (int i = 0; i < TRY; i++) {
          System.out.println("write~");
          int numWritten = writeTo(sc.socket(), i + 1);
          System.out.println("fini~");
          fini(numWritten);
          System.out.println("rollback done~ after 10 sec, retry");
          // System.out.println("fini done~ after 10 sec, rollback");

          try {
            Thread.sleep(10000);
          } catch (Exception e) {
          }

          /*
           * rollback();
           * System.out.println("rollback done~ after 10 sec, retry");
           * 
           * try { Thread.sleep(10000); } catch (Exception e) { }
           */
        }
      }
    } catch (UnknownHostException e) {
      // TODO Auto-generated catch block
      e.printStackTrace();
    } catch (IOException e) {
      // TODO Auto-generated catch block
      e.printStackTrace();
    } catch (SecurityException e) {
      // TODO Auto-generated catch block
      e.printStackTrace();
    } catch (NoSuchMethodException e) {
      // TODO Auto-generated catch block
      e.printStackTrace();
    }

    if (sc != null) {
      try {
        sc.socket().close();
      } catch (IOException e) {
        e.printStackTrace();
      }
    }
  }

  private static void removeLog() throws IOException, SecurityException, NoSuchMethodException {
    String pipeKey = "tabletName";

    Object[][] params = new Object[3][];
    params[0] = new Object[] { pipeKey };
    params[1] = new Object[] { pipeKey };
    params[2] = new Object[] { pipeKey };

    NConfiguration conf = new NConfiguration();
    Method method = ChangeLogServerIF.class.getMethod("removeLog", String.class);
    //NRPC.call(method, params, rpcAddresses, conf);
  }

  private static void rollback() throws IOException, SecurityException, NoSuchMethodException {
    String pipeKey = "tabletName";

    Object[][] params = new Object[3][];
    params[0] = new Object[] { pipeKey };
    params[1] = new Object[] { pipeKey };
    params[2] = new Object[] { pipeKey };

    NConfiguration conf = new NConfiguration();
    Method method = ChangeLogServerIF.class.getMethod("rollback", String.class);
    //NRPC.call(method, params, rpcAddresses, conf);
  }

  private static void fini(int numWritten) throws SecurityException, NoSuchMethodException,
      IOException {
    String pipeKey = "tabletName";

    Object[][] params = new Object[3][];
    params[0] = new Object[] { pipeKey, numWritten };
    params[1] = new Object[] { pipeKey, numWritten };
    params[2] = new Object[] { pipeKey, numWritten };

    NConfiguration conf = new NConfiguration();
    Method method = ChangeLogServerIF.class.getMethod("fini", String.class, int.class);
    //Object[] retList = NRPC.call(method, params, rpcAddresses, conf);
  }

  private static int writeTo(Socket socket, int seq) {
    try {
      byte[] txId = "txId1234567890".getBytes();
      byte[] bytes = TEST_STRING.getBytes();
      System.out.println("write bytes length : " + bytes.length);
      DataOutputStream dos = new DataOutputStream(socket.getOutputStream());

      dos.writeInt(seq); // seq number
      dos.writeInt(txId.length);
      dos.write(txId); // txid

      dos.write(bytes);
      dos.flush();
      System.out.println("write done");
      return bytes.length;
    } catch (IOException e) {
      e.printStackTrace();
    }
    return -1;
  }

  private static boolean connectPipeTo(Socket s) throws IOException {
    byte[] MAGIC_KEY = new byte[] { 0xC, 0xA, 0xF, 0xE, 0xB, 0xA, 0xB, 0xE };
    ByteBuffer buf = build();
    byte[] array = new byte[buf.limit()];
    System.arraycopy(buf.array(), 0, array, 0, array.length);

    DataOutputStream dos = new DataOutputStream(s.getOutputStream());
    dos.write(MAGIC_KEY);
    dos.writeInt(0);
    dos.writeInt(array.length);
    dos.write(array);
    dos.flush();

    DataInputStream dis = new DataInputStream(s.getInputStream());
    byte[] key = new byte[MAGIC_KEY.length];
    dis.read(key);
    System.out.println("MAGIC KEY CHECK : " + Arrays.equals(MAGIC_KEY, key));
    System.out.println("OP CODE : " + dis.readInt());
    int length = dis.readInt();
    System.out.println("BODY LENGTH : " + length);

    byte[] body = new byte[length];
    dis.read(body);
    String ret = new String(body);
    System.out.println("body : " + ret);

    if (ret.equals("OK")) {
      return true;
    }

    return false;
  }

  private static ByteBuffer build() throws IOException {
    ByteBuffer buf = ByteBuffer.allocate(1024);

    buf.putLong(0);
    ChangeLogWritableUtils.writeString(buf, "tabletName");

    String[] ipAddressList = new String[3];
    rpcAddresses = new InetSocketAddress[3];

    for (int i = 0; i < ipAddressList.length; i++) {
      ipAddressList[i] = "127.0.0.1:" + (18000 + i);
      rpcAddresses[i] = new InetSocketAddress("127.0.0.1", (57001 + i));
    }

    ChangeLogWritableUtils.writeStringArray(buf, ipAddressList);

    buf.flip();
    return buf;
  }

  private static void connect(SocketChannel sc) throws IOException {
    ByteBuffer headerBuf = ByteBuffer.allocate(16);
    ByteBuffer buf = build();

    byte[] MAGIC_KEY = new byte[] { 0xC, 0xA, 0xF, 0xE, 0xB, 0xA, 0xB, 0xE };

    try {
      headerBuf.put(MAGIC_KEY);
      headerBuf.putInt(0);
      headerBuf.putInt(buf.position());
      headerBuf.flip();
      buf.flip();
      sc.write(new ByteBuffer[] { headerBuf, buf });

      headerBuf.clear();
      buf.clear();

      sc.read(new ByteBuffer[] { headerBuf, buf });

      System.out.println("readFromNext pos : " + buf.position());
      System.out.println("readFromNext limit : " + buf.limit());

      buf.flip();
      byte[] bufArray = new byte[buf.limit()];
      buf.get(bufArray);
      System.out.println(buf.limit());
      System.out.println(new String(bufArray));
      buf.rewind();
    } catch (IOException e) {
      // TODO Auto-generated catch block
      e.printStackTrace();
    }
  }

  static final String TEST_STRING = "As recently as the 1990s, studies showed that most people preferred getting information from other people rather than from information retrieval (IR) systems. Of course, in that time period, most people also used human travel agents to book their travel. However, during the last decade, relentless optimization of information retrieval effectiveness has driven web search engines to new quality levels at which most people are satisfied most of the time, and web search has become a standard and often preferred source of information finding. For example, the 2004 Pew Internet Survey (Fallows 2004) found that “92% of Internet users say the Internet is a good place to go for getting everyday information.” To the surprise of many, the field of information retrieval has moved from being a primarily academic discipline to being the basis underlying most people’s preferred means of information access. This book presents the scientific underpinnings of this field, at a level accessible to graduate students as well as advanced undergraduates. Information retrieval did not begin with the Web. In response to various challenges of providing information access, the field of IR evolved to give principled approaches to searching various forms of content. The field began with scientific publications and library records but soon spread to other forms of content, particularly those of information professionals, such as journalists, lawyers, and doctors. Much of the scientific research on IR has occurred in these contexts, and much of the continued practice of IR deals with providing access to unstructured information in various corporate and governmental domains, and this work forms much of the foundation of our book.";
}
